package com.atliuzu.app.dws;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import com.atliuzu.app.func.DimAsyncFunction;
import com.atliuzu.bean.CommentBean;
import com.atliuzu.utils.ClickHouseUtil;
import com.atliuzu.utils.DateFormatUtil;
import com.atliuzu.utils.MyKafkaUtil;
import org.apache.flink.api.common.eventtime.SerializableTimestampAssigner;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.functions.FlatMapFunction;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.common.functions.ReduceFunction;
import org.apache.flink.streaming.api.datastream.AsyncDataStream;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.KeyedStream;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.windowing.AllWindowFunction;
import org.apache.flink.streaming.api.functions.windowing.WindowFunction;
import org.apache.flink.streaming.api.windowing.assigners.TumblingEventTimeWindows;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
import org.apache.flink.util.Collector;

import java.time.Duration;
import java.util.concurrent.TimeUnit;

public class DwsInteractionCourseCommentWindow {
    public static void main(String[] args) throws Exception {
        //todo 1.获取执行环境
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment().setParallelism(1);

        //todo 2.读取kfk DWD层评价主题数据，并创建流
        String topic = "dwd_interaction_comment";
        String groupId = "course_comment_window";

        DataStreamSource<String> kfkDS = env.addSource(MyKafkaUtil.getFlinkKafkaConsumer(topic, groupId));

        //todo 3.将流转化为json，同时过滤脏数据
        SingleOutputStreamOperator<JSONObject> jsonObjDS = kfkDS.flatMap(new FlatMapFunction<String, JSONObject>() {
            @Override
            public void flatMap(String value, Collector<JSONObject> out) throws Exception {
                try {
                    JSONObject jsonObject = JSON.parseObject(value);
                    out.collect(jsonObject);
                } catch (Exception ignored) {
                }
            }
        });


        //todo 4.将数据转化为javabean，并提取时间戳生成watermark
        SingleOutputStreamOperator<CommentBean> commentDS = jsonObjDS.map(new MapFunction<JSONObject, CommentBean>() {
            @Override
            public CommentBean map(JSONObject val) throws Exception {
                //定义好评人数
                Long goodUser = 0L;
                //定义评价星数
                Long stars;


                //判断是否为好评
                if (val.getString("review_stars").equals("5")) {
                    goodUser = 1L;
                }

                //获取评价星数
                stars = Long.parseLong(val.getString("review_stars"));

                //将数据输出为JavaBean
                return CommentBean.builder()
                        .courseId(val.getString("course_id"))
                        .usersCt(1L)
                        .goodUsersCt(goodUser)
                        .startsCt(stars)
                        .ts(DateFormatUtil.toTs(val.getString("create_time"), true))
                        .build();
            }
        }).assignTimestampsAndWatermarks(WatermarkStrategy.<CommentBean>forBoundedOutOfOrderness(Duration.ofSeconds(2))
        .withTimestampAssigner(new SerializableTimestampAssigner<CommentBean>() {
            @Override
            public long extractTimestamp(CommentBean element, long recordTimestamp) {
                return element.getTs();
            }
        }));
        //commentDS.print("2>>>>>>>>>>>>");

        //todo 7.分组、开窗、聚合
        SingleOutputStreamOperator<CommentBean> reduceDS = commentDS.keyBy(CommentBean::getCourseId)
                .window(TumblingEventTimeWindows.of(Time.seconds(10)))
                .reduce(new ReduceFunction<CommentBean>() {
                    @Override
                    public CommentBean reduce(CommentBean value1, CommentBean value2) throws Exception {
                        value1.setUsersCt(value1.getUsersCt() + value2.getUsersCt());
                        value1.setGoodUsersCt(value1.getGoodUsersCt() + value2.getGoodUsersCt());
                        value1.setStartsCt(value1.getStartsCt() + value2.getStartsCt());
                        return value1;
                    }
                }, new WindowFunction<CommentBean, CommentBean, String, TimeWindow>() {
                    @Override
                    public void apply(String s, TimeWindow window, Iterable<CommentBean> input, Collector<CommentBean> out) throws Exception {
                        CommentBean next = input.iterator().next();

                        next.setTs(System.currentTimeMillis());
                        next.setEdt(DateFormatUtil.toYmdHms(window.getEnd()));
                        next.setStt(DateFormatUtil.toYmdHms(window.getStart()));

                        out.collect(next);
                    }
                });

        //todo 8.关联维表补充信息
        SingleOutputStreamOperator<CommentBean> DSWithCourseName = AsyncDataStream.unorderedWait(reduceDS,
                new DimAsyncFunction<CommentBean>("DIM_COURSE_INFO") {
                    @Override
                    public String getKey(CommentBean input) {
                        return input.getCourseId();
                    }

                    @Override
                    public void join(CommentBean input, JSONObject dimInfo) {
                        input.setCourseName(dimInfo.getString("COURSE_NAME"));
                    }
                }, 100, TimeUnit.SECONDS);

        //TODO 8.将数据写出到ClickHouse
        //打印测试
        DSWithCourseName.print("1>>>>>>>>>>>>");
        DSWithCourseName.addSink(ClickHouseUtil.getJdbcSink("insert into dws_interaction_course_comment_window values(?,?,?,?,?,?,?,?)"));

        //TODO 9.启动任务
        env.execute("DwsInteractionCourseCommentWindow");


    }
}
